<?php

namespace Xin\EasyQueue\Hyperf\AsyncQueue\Driver;

use App\Supports\Logger\Log;
use Hyperf\AsyncQueue\Driver\ChannelConfig;
use Hyperf\AsyncQueue\Driver\RedisDriver;
use Hyperf\AsyncQueue\Exception\InvalidQueueException;
use Hyperf\AsyncQueue\JobInterface;
use Hyperf\AsyncQueue\JobMessage;
use Hyperf\AsyncQueue\Message;
use Hyperf\AsyncQueue\MessageInterface;
use Hyperf\Contract\StdoutLoggerInterface;
use Psr\Container\ContainerInterface;
use Psr\Log\LoggerInterface;

class RedisMultiQueueDriver extends RedisDriver
{
    /**
     * @var array<ChannelConfig>
     */
    protected array $secondaryChannels;

    /**
     * @var ChannelConfig|null
     */
    protected ChannelConfig|null $currentUseChanel = null;

    /**
     * @var LoggerInterface
     */
    private mixed $logger;

    /**
     * @param ContainerInterface $container
     * @param array $config
     */
    public function __construct(ContainerInterface $container, $config)
    {
        parent::__construct($container, $config);
        $this->logger = $this->container->get(StdoutLoggerInterface::class);
        $this->logger = Log::get();

        $this->currentUseChanel = $this->channel;

        $secondaryChannels = $config['secondary_channels'] ?? '';
        $secondaryChannels = is_array($secondaryChannels) ? $secondaryChannels : explode(',', $secondaryChannels);
        $secondaryChannels = array_filter($secondaryChannels);

        $this->secondaryChannels = array_map(function ($channelName) {
            return make(ChannelConfig::class, ['channel' => $channelName]);
        }, $secondaryChannels);
    }

    /**
     * @return array
     * @throws \RedisException
     */
    public function pop(): array
    {
        $channels = $this->getChannels();
        $channelCount = count($channels);
        /** @var ChannelConfig $channel */
        foreach ($channels as $channel) {
            $this->move($channel->getDelayed(), $channel->getWaiting());
            $this->move($channel->getReserved(), $channel->getTimeout());

            $res = $this->redis->brPop($channel->getWaiting(), $this->timeout);
            $this->logger->info("read queue({$channelCount}) task by:" . $channel->getChannel());
            if (!isset($res[1])) {
                continue;
//                return [false, null];
            }

            $data = $res[1];
            $message = $this->packer->unpack($data);
            if (!$message) {
                continue;
//                return [false, null];
            }

            $this->logger->info("read queue task by:" . $channel->getChannel());
            $this->currentUseChanel = $channel;
            $this->redis->zadd($channel->getReserved(), time() + $this->handleTimeout, $data);

            return [$data, $message];
        }

        $this->currentUseChanel = $this->channel;


        return [false, null];
    }

    public function delete(JobInterface $job): bool
    {
        if (class_exists(JobMessage::class)) {
            $message = \Hyperf\Support\make(JobMessage::class, [$job]);
        } else {
            $message = \make(Message::class, [$job]);
        }

        $data = $this->packer->pack($message);

        return (bool)$this->redis->zRem($this->currentUseChanel->getDelayed(), $data);
    }

    public function ack($data): bool
    {
        return $this->remove($data);
    }

    public function fail($data): bool
    {
        if ($this->remove($data)) {
            return (bool)$this->redis->lPush($this->currentUseChanel->getFailed(), $data);
        }
        return false;
    }

    public function reload(string $queue = null): int
    {
        $num = 0;

        $channels = $this->getChannels();
        /** @var ChannelConfig $channelConfig */
        foreach ($channels as $channelConfig) {
            $channel = $channelConfig->getFailed();
            if ($queue) {
                if (!in_array($queue, ['timeout', 'failed'])) {
                    throw new InvalidQueueException(sprintf('Queue %s is not supported.', $queue));
                }

                $channel = $channelConfig->get($queue);
            }

            while ($this->redis->rpoplpush($channel, $channelConfig->getWaiting())) {
                ++$num;
            }
        }

        return $num;
    }

    public function flush(string $queue = null): bool
    {
        return $this->arraySome($this->getChannels(), function ($channelConfig) use ($queue) {
            /** @var ChannelConfig $channelConfig */
            $channel = $channelConfig->getFailed();
            if ($queue) {
                $channel = $channelConfig->get($queue);
            }

            return (bool)$this->redis->del($channel);
        });
    }

    public function info(): array
    {
        return [
            'waiting' => $this->redis->lLen($this->channel->getWaiting()),
            'delayed' => $this->redis->zCard($this->channel->getDelayed()),
            'failed' => $this->redis->lLen($this->channel->getFailed()),
            'timeout' => $this->redis->lLen($this->channel->getTimeout()),
        ];
    }

    protected function retry(MessageInterface $message): bool
    {
        return $this->arraySome($this->getChannels(), function ($channelConfig) use ($message) {
            /** @var ChannelConfig $channelConfig */
            $data = $this->packer->pack($message);

            $delay = time() + $this->getRetrySeconds($message->getAttempts());

            return $this->redis->zAdd($channelConfig->getDelayed(), $delay, $data) > 0;
        });
    }

    protected function getRetrySeconds(int $attempts): int
    {
        if (!is_array($this->retrySeconds)) {
            return $this->retrySeconds;
        }

        if (empty($this->retrySeconds)) {
            return 10;
        }

        return $this->retrySeconds[$attempts - 1] ?? end($this->retrySeconds);
    }

    /**
     * Remove data from reserved queue.
     * @param string $data
     */
    protected function remove($data): bool
    {
        return $this->redis->zrem($this->currentUseChanel->getReserved(), $data) > 0;
    }

    /**
     * @return array
     */
    protected function getChannels(): array
    {
        return [$this->channel, ...$this->secondaryChannels];
    }

    protected function arraySome($array, $callback)
    {
        $result = [];
        foreach ($array as $key => $item) {
            $result = $callback($item, $key);
        }
        return count(array_filter($result)) > 0;
    }
}
